package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job05_累计GMV看板 {

    public static void main(String[] args) {

        // 创建环境
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        // 开启窗口的提前触发器（trigger）
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1000 ms");


        // 创建业务库中的oms_order表的cdc连接器表
        tenv.executeSql(
                        " create table oms_order_mysql_cdc(                 \n "+
                        "   id           bigint,                            \n "+
                        " 	total_amount decimal(10,2),                     \n "+
                        " 	pay_amount   decimal(10,2),                     \n "+
                        " 	create_time  timestamp(3),                      \n "+
                        " 	modify_time  timestamp(3),                      \n "+
                        " 	rt as modify_time,                              \n "+
                        " 	watermark for rt as rt - interval '0' second,   \n "+
                        "   primary key (id) not enforced                   \n "+
                        " )WITH (                                           \n "+
                        "     'connector' = 'mysql-cdc',                    \n "+
                        "     'hostname' = 'doitedu',                       \n "+
                        "     'port' = '3306',                              \n "+
                        "     'username' = 'root',                          \n "+
                        "     'password' = 'root',                          \n "+
                        "     'database-name' = 'realtimedw',               \n "+
                        "     'table-name' = 'oms_order'                    \n "+
                        " )                                                 \n "
        );

        // 创建mysql目标表的jdbc连接器表
        tenv.executeSql(
                        " create table dash_05_cumulate_gmv_mysql(        \n "+
                        "                                                 \n "+
                        "   `window_start` timestamp(3) ,                  \n "+
                        "   `window_end` timestamp(3) ,                    \n "+
                        "   `total_amount` decimal(10,2) ,                \n "+
                        "   `pay_amount` decimal(10,2) ,                  \n "+
                        "   `discount_amount` decimal(10,2) ,             \n "+
                        "   primary key (window_start) not enforced       \n "+
                        " ) with (                                        \n "+
                        "    'connector' = 'jdbc',                        \n "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/doit43',  \n "+
                        "    'table-name' = 'dash_05_cumulate_gmv',       \n "+
                        "    'username' = 'root'      ,                   \n "+
                        "    'password' = 'root'                          \n "+
                        " )                                               \n "
        );



        // 核心逻辑：累计窗口
        tenv.executeSql(
                        " INSERT INTO  dash_05_cumulate_gmv_mysql                     "+
                        " WITH tmp AS (                                                "+
                        "     SELECT                                                   "+
                        "         *                                                    "+
                        "     from  oms_order_mysql_cdc                                "+
                        "     where                                                    "+
                        "     date_format(create_time,'yyyy-MM-dd') =  current_date    "+
                        " )                                                            "+
                        "                                                              "+
                        " SELECT                                                       "+
                        "     TUMBLE_START(rt,INTERVAL '24' HOUR),                    "+
                        "     TUMBLE_END(rt,INTERVAL '24' HOUR),                      "+
                        " 	sum(total_amount) as total_amount,                         "+
                        " 	sum(pay_amount)   as pay_amount,                           "+
                        " 	sum(total_amount - pay_amount)   as discount_amount        "+
                        " from tmp                                                     "+
                        " group by                                                     "+
                        "     TUMBLE(rt,INTERVAL '24' HOUR)                           "
        );




    }

}
